package sg.edu.nus.iss.stockex;

import java.io.IOException;
import java.util.Iterator;
import java.util.LinkedList;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.Reducer.Context;

public class RelativeStrengthIndexReducerWithoutRunningWindow extends Reducer<CodeDateKey, TimestampPriceDataPoint, Text, Text> {

	static long DAY_IN_MS = 24 * 60 * 60 * 1000;

	@Override
	public void reduce(CodeDateKey key, Iterable<TimestampPriceDataPoint> values,
			Context context) throws IOException, InterruptedException {


		Text outKey = new Text();
		Text outVal = new Text();

		int countGain = 0;
		int countLoss = 0;
		double totalGain = 0;
		double totalLoss = 0;

		TimestampPriceDataPoint lastDataPoint = null;

		for(TimestampPriceDataPoint dataPoint: values)
		{
			if(dataPoint.getClosePrice()>dataPoint.getOpenPrice())
			{
				totalGain+=dataPoint.getClosePrice()-dataPoint.getOpenPrice();
				countGain++;
			} else{
				totalLoss += dataPoint.getOpenPrice()-dataPoint.getClosePrice();
				countLoss++;
			}
			lastDataPoint = dataPoint;
		}
		double averageGain = totalGain/countGain;
		double averageLoss = totalLoss/countLoss;
		double relativeStrength =  averageGain/averageLoss;
		double relativeStrengthIndex = 100 - 100/(1+relativeStrength);
		if (Double.isNaN(relativeStrengthIndex)){
			relativeStrengthIndex = 0;
		}

/*
		outKey.set("Code: " + key.getCode() + ", Date: "
				+ lastDataPoint.getDate());
		outVal.set("RSI: " + relativeStrengthIndex);
*/
		
		outKey.set(key.getCode() +  "\t" + lastDataPoint.getDate());
		outVal.set(""+relativeStrengthIndex);
		context.write(outKey, outVal);

	} // end reduce

}
